package com.queue.cli.handler;

import com.queue.cli.constant.Destination;
import com.queue.cli.constant.MessageType;
import com.queue.cli.listener.MessageData;
import com.queue.cli.listener.MessageListener;
import com.queue.common.message.Message;
import com.queue.common.protocol.MessageUtils;
import com.queue.common.protocol.Protocol;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;

import java.util.concurrent.TimeUnit;

/**
 * 订阅者的消息处理器
 */
public class SubscriberHandler extends SimpleChannelInboundHandler<Protocol> {
    /**
     * 订阅的主题
     */
    private String destination;

    /**
     * 订阅者唯一标识
     */
    private long subscriberId;

    /**
     * 监听器
     */
    private MessageListener messageListener;


    public SubscriberHandler(String destination,long subscriberId){
        this.destination = destination;
        this.subscriberId = subscriberId;
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        //向服务端注册该订阅者
        register(ctx.channel());
    }

    private void register(Channel channel) {
        new Message(subscriberId, MessageType.CONSUMER.getType(),null,this.destination, Destination.TOPIC.getDestination(), false,0, TimeUnit.MILLISECONDS);
    }

    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, Protocol protocol) throws Exception {
        //将字节数组转化成消息对象
        Message message = MessageUtils.reverse(protocol.getContent());
        if (message!=null){
            execListener(new MessageData(this, message.getContent()));
        }
    }

    private void execListener(MessageData data) {
        this.messageListener.getMessage(data);
    }
}
